package day03;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

public class KafkaSinkTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        Properties properties = new Properties();

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"s1.hadoop:9092");
        properties.put(ProducerConfig.ACKS_CONFIG,"all");//ACK消息应答机制
        properties.put(ProducerConfig.RETRIES_CONFIG,"3");//重试次数
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,"100");
        properties.put(ProducerConfig.LINGER_MS_CONFIG,"100");

       // ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG


        env.addSource(new SourceFunction<Tuple2<String,String>>() {
            Boolean flag =true;
            @Override
            public void run(SourceContext<Tuple2<String, String>> ctx) throws Exception {
                int i=0;
                while(flag){
                    ctx.collect(Tuple2.of("hainiu",i+""));
                    Thread.sleep(3000);
                    i++;
                }
            }

            @Override
            public void cancel() {
                flag=false;
            }
            //brokerList topic seria [partitions]
        }).addSink(new FlinkKafkaProducer010<Tuple2<String, String>>("topic_33", new SerializationSchema<Tuple2<String, String>>() {
            @Override
            public byte[] serialize(Tuple2<String, String> element) {
                return (element.f0+ " "+element.f1).getBytes();
            }
        },properties));

        env.execute();
    }
}
